package com.itlong.cloud.rabbitmq.consumer;

import com.itlong.cloud.push.message.GatewayService;
import org.apache.commons.lang.StringUtils;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * <desc>
 *      订阅消息
 * </desc>
 *
 * @return
 * @Author Zhu.yj
 * @createDate  2021/7/1
 */
@Component
public class WebSocketPushMessageListener implements MessageListener {

    private Logger logger = LoggerFactory.getLogger(WebSocketPushMessageListener.class);

    @Autowired
    private StringRedisTemplate redisTemplate;


    @Override
    public void onMessage(Message message, byte[] pattern) {
        RedisSerializer<String> valueSerializer = redisTemplate.getStringSerializer();
        String value = valueSerializer.deserialize(message.getBody());

        if (StringUtils.isNotBlank(value)) {
            logger.info("监听websocket消息---{}", value);
            //集群模式 推送消息
            Map<String, Channel> map = GatewayService.getChannels();
            for (String key : map.keySet()) {
                Channel obj = map.get(key);
                if (obj.isOpen() && obj.isConnected()) {
                    obj.write(new TextWebSocketFrame(value));
                } else {
                    GatewayService.removeGatewayChannel(key);
                }
            }
        }
    }

}